package com.fengye.revoker;

import com.fengye.model.RemoteRequest;
import com.fengye.model.RemoteResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * @Author fengye
 * @Date 2023/2/15 14:12
 **/
public class RevokerServiceCallable implements Callable<RemoteResponse> {

    private final static Logger logger = LoggerFactory.getLogger(RevokerServiceCallable.class);

    /** channel通道 */
    private Channel channel;
    /** 调用地址信息 */
    private InetSocketAddress inetSocketAddress;
    /** 服务请求 */
    private RemoteRequest request;

    public RevokerServiceCallable(InetSocketAddress inetSocketAddress, RemoteRequest request) {
        this.inetSocketAddress = inetSocketAddress;
        this.request = request;
    }

    public static RevokerServiceCallable of(InetSocketAddress inetSocketAddress, RemoteRequest request) {
        return new RevokerServiceCallable(inetSocketAddress, request);
    }

    @Override
    public RemoteResponse call() throws Exception {
        // 初始化返回结果容器，将本次调用的唯一标识作为Key存入
        RevokerResponseHolder.initResponseData(request.getUniqueKey());
        // 根据本地调用服务提供者地址获取对应的Netty通道channel队列
        ArrayBlockingQueue<Channel> blockingQueue = NettyChannelPoolFactory.channelPoolFactoryInstance().acquire(inetSocketAddress);
        try {
            // 如果channel为空，从队列中获取本次调用的Netty通道channel
            if (channel == null) {
                channel = blockingQueue.poll(request.getInvokeTimeout(), TimeUnit.MICROSECONDS);
            }

            // 如果获取的chaneel不可用，则重新获取一个,就是一定要有
            while (!channel.isOpen() || !channel.isActive() || !channel.isWritable()) {
                logger.warn("---------retry get new Channel from queue-----------");
                channel = blockingQueue.poll(request.getInvokeTimeout(), TimeUnit.MICROSECONDS);
                if (channel == null) {
                    logger.warn("---------retry get new Channel from new-----------");
                    // 若队列中没有可用的channel，则重新注册一个
                    channel = NettyChannelPoolFactory.channelPoolFactoryInstance().registerChannel(inetSocketAddress);
                }
            }

            // 将本次调用的信息写入Netty通道，发起异步调用
            ChannelFuture channelFuture = channel.writeAndFlush(request);
            channelFuture.syncUninterruptibly();
            // 从返回结果容器中获取返回结果，同时设置超时时间
            return RevokerResponseHolder.getResultValue(request.getUniqueKey(), request.getInvokeTimeout());
        } catch (Exception e) {
            logger.error("service invoke error.", e);
        } finally {
            // 本次调用结束，将channel重新释放到队列中，以便下次调用
            NettyChannelPoolFactory.channelPoolFactoryInstance().release(blockingQueue, channel, inetSocketAddress);
        }

        return null;
    }
}
